/*
 * Copyright (c) 2015. 速位中国
 */

package com.netty;


import com.controller.abstractcontroller.AbstractController;
import com.controller.abstractcontroller.IController;
import com.flatgen.LoginRequest;
import com.google.flatbuffers.FlatBufferBuilder;
import com.vo.PlayerInstance;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.ChannelInputShutdownEvent;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.NoSuchElementException;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * 逻辑分发处理
 * <p/>
 * 屏蔽掉I/O 逻辑处理日志输出
 */
@Component
public class ServerHandler extends ChannelInboundHandlerAdapter {

    public final Logger logger =
            LoggerFactory.getLogger(ServerHandler.class);

    /**
     * 记录服务器在线人数--连接数
     */
    private static final AtomicInteger countConnection = new AtomicInteger(0);


    /**
     *
     */
    private PlayerInstance playerInstance;


    /**
     * channel is active
     */
    private boolean active = false;//channel is active  ?

    /**
     * 读取数据超时次数
     */
    private int nettyReadTimeoutTimes;


    /**
     * @param nettyReadTimeoutTimes
     */
    protected ServerHandler(int nettyReadTimeoutTimes) {
        this.nettyReadTimeoutTimes = nettyReadTimeoutTimes;
    }


    /**
     * 添加监听器，在指定的时间发送消息
     * Calls {@link io.netty.channel.ChannelHandlerContext#fireChannelActive()} to forward
     * to the next {@link io.netty.channel.ChannelInboundHandler} in the {@link io.netty.channel.ChannelPipeline}.
     * <p/>
     * Sub-classes may override this method to change behavior.
     * sub_reactor
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        active = true;
        if (logger.isDebugEnabled()) countConnection.getAndIncrement();// +1
        playerInstance = new PlayerInstance();  //create new playerInstance instance
        playerInstance.setChannel(ctx.channel());
        if (logger.isDebugEnabled())
            logger.debug(ctx.channel() + "connect server... Concurrent  connection... " + countConnection.get());


        testflat(ctx);
    }


    /**
     * @param ctx
     */
    private void testflat(ChannelHandlerContext ctx) {

        for (int i=0;i!=3;++i){
            FlatBufferBuilder flatBufferBuilder = new FlatBufferBuilder();
            int offset = LoginRequest.createLoginRequest(flatBufferBuilder, 1, flatBufferBuilder.createString("石头哥哥"));
            flatBufferBuilder.finish(offset);
            ByteBuf byteBuf = AbstractController.buildBuf(1, flatBufferBuilder);
            ctx.channel().writeAndFlush(byteBuf);
        }
    }


    /**
     * 逻辑处理线程池 处理业务 ---- 如果线程处理不过来 那么 数据将会缓存在线程待处理的任务队列中
     * 这里线程串行化处理，效果不佳，根据实际的压测数据来处理：1.存在耗时的操作--脱离当前的业务处理线程池；
     * 2.采用队列来缓存待处理的消息 （当前方法直接I/0线程，在当前的session中采用队列缓存）
     * <p/>
     * 采用串行化 和对应的缓存队列来处理消息都可能出现等待--数据处理不过来
     * 所以根本的解决方案 1.尽量避免减小耗时操作 2.增加业务处理线程池大小增强处理能力；
     * <p/>
     * 底层游戏数据分发处理
     * // recycle in all cases
     * out.recycle();//  clear list object  a ,for gc work
     *
     * @param ctx
     * @param message
     */
    @SuppressWarnings("Duplicates")
    public void channelRead(ChannelHandlerContext ctx, Object message) {
        try {
            /**
             *  work logic
             */
            byte[] message_byte = (byte[]) message;
            if (active) {
                //获取 消息标识ID
                int ID = (message_byte[0x0] & 0xff) << 0x18 |
                        (message_byte[0x1] & 0xff) << 0x10 |
                        (message_byte[0x2] & 0xff) << 0x8 | message_byte[0x3] & 0xff;

                int length = message_byte.length - 0x4;//数据包长度

                // for flat buffer
                if (length != 0) {
                    IController.game_controllers[ID].DispatchFlatBuffer(message_byte, playerInstance);
                } else {      //only id
                    IController.game_controllers[ID].DispatchFlatBuffer(null, playerInstance);
                }

            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * Calls {@link io.netty.channel.ChannelHandlerContext#fireChannelInactive()} to forward
     * to the next {@link io.netty.channel.Channel} in the {@link io.netty.channel.ChannelPipeline}.
     * <p/>
     * Sub-classes may override this method to change behavior.
     * <p/>
     * 注意回收资源      read time out  will be call back this function
     * sub_reactor
     * <p/>
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        try {
            //flg channel unActive
            active = false;


            //close channel
            ctx.channel().close();


            if (logger.isDebugEnabled()) countConnection.decrementAndGet();// -1

            if (logger.isDebugEnabled())
                logger.debug(ctx.channel().remoteAddress() + " channelClosed , Concurrent  connection... " + countConnection.get());

            ChannelPipeline channelPipeline = ctx.channel().pipeline();

            if (null == channelPipeline) {
                // if null ,will be returned
                //always clear playerInstance object
//                try {
//                    if (playerInstance != null) this.playerInstance.saveAndDestroy();//
//                } catch (Exception e) {
//                    if (logger.isDebugEnabled()) logger.error(e.getMessage());
//                }
                playerInstance = null;    //for gc
                return;
            }
            //remove pipeline object
            while (channelPipeline.last() != null) {
                channelPipeline.removeLast();
            }
            //delete
            if (ctx.isRemoved()) {
                ctx.close();
                // Set to null so the GC can collect them directly
                //clear and store data!  channel closed!------
                try {
                    //if (playerInstance != null) this.playerInstance.saveAndDestroy();//
                } catch (Exception e) {
                    if (logger.isDebugEnabled()) logger.error(e.getMessage());
                }
                playerInstance = null;    //for gc
            }
        } catch (NoSuchElementException ee) {
            ee.printStackTrace();
            //do nothing
        }
    }


    /**
     * 读取数据超时   --->  断定连接断开  ----> 释放对应的socket连接
     * <p/>
     * 心跳处理
     * 链路读写超时处理
     *
     * @param ctx
     * @param evt
     * @throws Exception sub_reactor                          ChannelInputShutdownEvent.INSTANCE
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        try {
            Channel channel = ctx.channel();
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent e = (IdleStateEvent) evt;
                if (e.state() == IdleState.READER_IDLE) {
                    channel.close();  //call back channelInactive(ChannelHandlerContext ctx)
                    if (logger.isDebugEnabled()) logger
                            .debug(channel.remoteAddress() + "---No data was received for a while ,read time out... ...");
                } //	  because we are attaching  more importance to the read_idle time(timeout to rec)
                else if (e.state() == IdleState.WRITER_IDLE) { // No data was sent for a while.
                    channel.close();
                    if (logger.isDebugEnabled()) logger
                            .debug(channel.remoteAddress() + "---No data was sent for a while.write time out... ...");
                }
            } else if (evt instanceof ChannelInputShutdownEvent) {     //处理远程主机强制关闭连接
                channel.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * Sub-classes may override this method to change behavior.
     * sub_reactor
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

}
